package com.jn.rxjava.views;

import android.annotation.SuppressLint;

import com.jn.rxjava.R;
import com.jn.rxjava.base.BaseActivity;
import com.jn.rxjava.utils.LogUtils;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.android.schedulers.AndroidSchedulers;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.ObservableSource;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.BiFunction;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.functions.Predicate;
import io.reactivex.rxjava3.observables.GroupedObservable;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class MainActivity extends BaseActivity {
    private Observable<String> mObservable;
    private Observable<Integer> mObservable2;
    private Observer<String> mObserver;
    private Observer<String> mObserver2;
    private Observer<Integer> mObserver3;
    private Observer<Integer> mObserver4;
    int z = 1;

    @Override
    public int getLayoutId() {
        return R.layout.activity_main;
    }

    @Override
    public void init() {
        /*initObservable();
        initObserver();
        mObservable.subscribe(mObserver);
        mObservable.subscribe(mObserver2);*/
        //testFlowable();

        //testInterval();
        //testIntervalRange();
        //testTimer();

        //testBuffer();
        //testGroupBy();
        //testCast();

        //testFilter();
        //testDistinct();
        //testTake();
        //testSkip();
        //testSkipLast();
        //testElementAt();
        //testTakeLast();
        //testTakeWhile();
        //testSkipWhile();
        //testOfType();

        //testMerge();
        //testConcat();
        //testZip();
        //testCombineLatest();
        //testStartWith();

        //testMap();
        //testFlatMap();
        //testConcatMap();
    }

    private void initObservable() {
        //create：手动创建一个 Observable。
        mObservable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                //重写subscribe()定义需要发送的事件
                //ObservableEmitter有三种发射方法，onNext()、onComplete()、onError()
                //onNext()发送事件，可以无限调用，所有的发射信息观察者都可以收得到
                emitter.onNext("电视剧第一集");
                emitter.onNext("电视剧第二集");
                emitter.onNext("电视剧第三集");
                //onComplete()、onError()是互斥的，观察者只会收到一个
                //onComplete()完成事件发送，可以重复调用，但是观察者只会接收一次
                emitter.onComplete();
                //onError()异常发送，不可以重复调用
                //emitter.onError(new IllegalArgumentException("TV Play Exception!"));
            }
        });

        //just:创建一个发射指定数据项的 Observable
        //mObservable = Observable.just("电视剧第一集", "电视剧第二集", "电视剧第三集");
    }


    /**
     * 背压
     */
    private void testFlowable() {
        Flowable.create(new FlowableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(@NonNull FlowableEmitter<Integer> emitter) throws Throwable {
                        for (int i = 0; i < 1000; i++) {
                            LogUtils.d("asdf", "========send========" + i);
                            emitter.onNext(i);
                        }
                    }
                }, BackpressureStrategy.DROP)
                .observeOn(AndroidSchedulers.mainThread())// 指定 最后拿到数据操，解析，显示发生在主线程
                .subscribeOn(Schedulers.computation())// 指定 网络请求耗时操作发生在子线程;
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onSubscribe(Subscription s) {
                        LogUtils.d("asdf", "========flow=====onSubscribe===");
                        s.request(1000);
                    }

                    @Override
                    public void onNext(Integer s) {
                        LogUtils.d("asdf", "========flow========" + s);
                    }

                    @Override
                    public void onError(Throwable t) {
                        LogUtils.d("asdf", "========flow=====onError===");
                    }

                    @Override
                    public void onComplete() {
                        LogUtils.d("asdf", "========flow=====onComplete===");
                    }
                });
    }

    //interval： 创建一个 Observable，定期发射一个递增的长整型数值
    private void testInterval() {
        Observable.interval(1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {
                        LogUtils.d("asdf", "========interval========" + aLong);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private void testIntervalRange() {
        // 从 10 开始，发射 3 个事件，初始延迟 0 秒，间隔 1 秒
        Observable.intervalRange(10, 3, 0, 1, TimeUnit.SECONDS)
                .subscribe(new Observer<Long>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Long aLong) {
                        LogUtils.d("asdf", "========intervalRange========" + aLong);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //timer： 创建一个 Observable，在指定延迟后发射一个数据项
    private void testTimer() {
        Observable.timer(5, TimeUnit.SECONDS)
                .subscribe(new Consumer<Long>() {//Consumer可理解为简化版的Observer
                    @Override
                    public void accept(Long aLong) throws Throwable {
                        LogUtils.d("asdf", "========timer========" + aLong);
                    }
                });
    }

    //buffer 用于将数据项分组为列表，并以列表的形式发射
    private void testBuffer() {
        Observable.just(1, 2, 3, 4, 5, 6, 7)
                .buffer(2)//每组2条数据
                .subscribe(new Consumer<List<Integer>>() {
                    @Override
                    public void accept(List<Integer> integers) throws Throwable {
                        for (int i = 0; i < integers.size(); i++) {
                            LogUtils.d("asdf", "=======buffer======第" + z + "组,第(" + i + ")条数据=" + integers.get(i));
                        }
                        z++;
                    }
                });
    }

    //groupBy ：将 Observable 数据项按某个标准进行分组，然后发射多个子 Observable，每个子 Observable 包含一组具有相同标准的数据项。
    private void testGroupBy() {
        Observable.just(-2, -1, 1, 2, 3, 4, 5, 6)
                .groupBy(new Function<Integer, String>() {
                    @Override
                    public String apply(Integer integer) throws Throwable {
                        if (integer > 0) {
                            return "正数";
                        }
                        return "负数";
                    }
                }).subscribe(new Observer<GroupedObservable<String, Integer>>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull GroupedObservable<String, Integer> stringIntegerGroupedObservable) {
                        LogUtils.d("asdf", "========groupBy getkey========" + stringIntegerGroupedObservable.getKey());
                        stringIntegerGroupedObservable.subscribe(new Consumer<Integer>() {
                            @Override
                            public void accept(Integer integer) throws Throwable {
                                LogUtils.d("asdf", "========groupBy========" + integer);
                            }
                        });
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //cast 将上游事件强制转换成指定类型
    private void testCast() {
        Observable.just(1, 2, 3)
                .cast(Number.class) // 转换为 Number 类型
                .subscribe(new Consumer<Number>() {
                    @Override
                    public void accept(Number number) throws Throwable {
                        LogUtils.d("asdf", "========cast========" + number);
                    }
                });
    }

    //filter：过滤掉不满足条件的数据项。
    @SuppressLint("CheckResult")
    private void testFilter() {
        Observable.just(-2, -1, 1, 2, 3, 4, 5, 6)
                .filter(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Throwable {
                        return integer > 0;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========filter========" + integer);
                    }
                });
    }

    //distinct：过滤掉重复的数据项
    //distinctUntilChanged： 用于过滤掉连续重复的数据项，只保留第一次出现的数据项
    @SuppressLint("CheckResult")
    private void testDistinct() {
        Observable.just(1, 2, 2, 1, 3, 4, 5, 5, 6)
                .distinct()
                //.distinctUntilChanged()
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========distinct========" + integer);
                    }
                });
    }

    //take：仅发射前 N 个数据项。
    @SuppressLint("CheckResult")
    private void testTake() {
        Observable.just(1, 2, 2, 1, 3, 4, 5, 5, 6)
                .take(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========take========" + integer);
                    }
                });
    }

    //skip：跳过前 N 个数据项。
    @SuppressLint("CheckResult")
    private void testSkip() {
        Observable.just(1, 2, 2, 1, 3, 4, 5, 5, 6)
                .skip(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========skip========" + integer);
                    }
                });
    }

    //skipLast：跳过后 N 个数据项。
    @SuppressLint("CheckResult")
    private void testSkipLast() {
        Observable.just(1, 2, 2, 1, 3, 4, 5, 5, 6)
                .skipLast(3)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========skip========" + integer);
                    }
                });
    }

    //elementAt： 用于发射指定索引位置的数据项，忽略其他数据项。
    @SuppressLint("CheckResult")
    private void testElementAt() {
        Observable.just(1, 2, 2, 1, 3, 4, 5, 5, 6)
                //.elementAt(4)
                .elementAtOrError(20)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========elementAt========" + integer);
                    }
                });
    }

    //takeLast： 用于仅发射最后 N 个数据项，忽略前面的数据项。
    @SuppressLint("CheckResult")
    private void testTakeLast() {
        Observable.just(1, 2, 2, 1, 3, 4, 5, 5, 6)
                .takeLast(4)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========takeLast========" + integer);
                    }
                });
    }

    //takeWhile： 从事件流开头开始，持续保留满足条件的事件，直到遇到第一个不满足条件的事件，之后的所有事件都会被丢弃。
    @SuppressLint("CheckResult")
    private void testTakeWhile() {
        Observable.just(1, 2, 2, 1, 3, 4, 1, 5, 2, 6)
                .takeWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Throwable {
                        return integer < 3;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========takeWhile========" + integer);
                    }
                });
    }

    //skipWhile： 从事件流开头开始，持续跳过满足条件的事件，直到遇到第一个不满足条件的事件，之后的所有事件都会被保留。
    @SuppressLint("CheckResult")
    private void testSkipWhile() {
        Observable.just(1, 2, 2, 1, 3, 4, 1, 5, 2, 6)
                .skipWhile(new Predicate<Integer>() {
                    @Override
                    public boolean test(Integer integer) throws Throwable {
                        return integer < 3;
                    }
                }).subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========skipWhile========" + integer);
                    }
                });
    }

    //ofType： 只保留符合类型的数据。
    @SuppressLint("CheckResult")
    private void testOfType() {
        Observable.just(1, "A", 3, "B")
                .ofType(Integer.class)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Throwable {
                        LogUtils.d("asdf", "========ofType========" + integer);
                    }
                });
    }

    private void testMerge() {
        // 创建两个带延迟的 Observable（模拟异步执行速度差异）
        Observable<Integer> obs1 = Observable.just(1, 2, 3, 4)
                .delay(100, TimeUnit.MILLISECONDS); // 延迟100ms
        Observable<String> obs2 = Observable.just("A", "B", "C", "D")
                .delay(50, TimeUnit.MILLISECONDS);  // 延迟50ms
        Observable
                .merge(obs1, obs2)//最多支持4个
                //.mergeArray(obs1, obs2)//支持多个
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Object o) {
                        LogUtils.d("asdf", "========merge========" + o.toString());
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private void testConcat() {
        // 创建两个带延迟的 Observable（模拟异步执行速度差异）
        Observable<Integer> obs1 = Observable.just(1, 2, 3, 4)
                .delay(100, TimeUnit.MILLISECONDS); // 延迟100ms
        Observable<String> obs2 = Observable.just("A", "B", "C", "D")
                .delay(50, TimeUnit.MILLISECONDS);  // 延迟50ms
        Observable
                .concat(obs1, obs2)//最多支持4个
                //.concatArray(obs1, obs2)//支持多个
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Object o) {
                        LogUtils.d("asdf", "========concat========" + o.toString());
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private void testZip() {
        Observable<Integer> obs1 = Observable.just(1, 2, 3, 4, 5, 6)
                .delay(100, TimeUnit.MILLISECONDS); // 延迟100ms
        Observable<String> obs2 = Observable.just("A", "B", "C", "D")
                .delay(50, TimeUnit.MILLISECONDS);  // 延迟50ms
        Observable.zip(obs2, obs1, new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Throwable {
                return s + " " + integer;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull String s) {
                LogUtils.d("asdf", "========zip========" + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private void testCombineLatest() {
        Observable<String> oa = Observable.interval(2, TimeUnit.SECONDS)
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Throwable {
                        return aLong + "A";
                    }
                });
        Observable<String> ob = Observable.interval(3, TimeUnit.SECONDS)
                .map(new Function<Long, String>() {
                    @Override
                    public String apply(Long aLong) throws Throwable {
                        return aLong + "B";
                    }
                });
        Observable.combineLatest(oa, ob, new BiFunction<String, String, String>() {
            @Override
            public String apply(String s, String s2) throws Throwable {
                return s + " ==== " + s2;
            }
        }).subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                LogUtils.d("asdf", "========combineLatest========" + s);
            }
        });
    }

    private void testStartWith() {
        Observable.just("B", "C")
                .startWith(Observable.just("A", "a")) // 前缀：A
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Throwable {
                        LogUtils.d("asdf", "========startWith========" + s);
                    }
                });
    }

    private void testMap() {
        Observable.just("电视剧第一集", "电视剧第二集", "电视剧第三集")
                .map(new Function<String, Integer>() {
                    @Override
                    public Integer apply(String s) throws Throwable {
                        int i = 0;
                        switch (s) {
                            case "电视剧第一集":
                                i = 1;
                                break;
                            case "电视剧第二集":
                                i = 2;
                                break;
                            case "电视剧第三集":
                                i = 3;
                                break;
                        }
                        return i;
                    }
                }).subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        LogUtils.d("asdf", "===============testMap==== " + integer);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    private void testFlatMap() {
        //range：创建一个发射特定整数范围的 Observable。
        mObservable2 = Observable.range(1, 10)
                .flatMap(new Function<Integer, ObservableSource<? extends Integer>>() {
                    @Override
                    public ObservableSource<? extends Integer> apply(Integer integer) throws Throwable {
                        List<Integer> list = new ArrayList<>();
                        list.add(integer * 10);
                        list.add(integer * 100);
                        list.add(integer * 1000);
                        return Observable.fromIterable(list);//fromIterable：从一个 Iterable 对象（如 List 或 Set）中创建一个 Observable。
                    }
                });
        mObserver3 = new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "3 onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                LogUtils.d("asdf", "3 onNext : " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "3 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "3 onComplete");
            }
        };

        mObserver4 = new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "4 onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                LogUtils.d("asdf", "4 onNext : " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "4 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "4 onComplete");
            }
        };
        mObservable2.subscribe(mObserver3);
        mObservable2.subscribe(mObserver4);
    }

    private void testConcatMap() {
        //range：创建一个发射特定整数范围的 Observable。
        mObservable2 = Observable.range(1, 10)
                .concatMap(new Function<Integer, ObservableSource<? extends Integer>>() {
                    @Override
                    public ObservableSource<? extends Integer> apply(Integer integer) throws Throwable {
                        List<Integer> list = new ArrayList<>();
                        list.add(integer + 10);
                        list.add(integer * 100);
                        list.add(integer + 1000);
                        return Observable.fromIterable(list);//fromIterable：从一个 Iterable 对象（如 List 或 Set）中创建一个 Observable。
                    }
                });
        mObserver3 = new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "3 onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                LogUtils.d("asdf", "3 onNext : " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "3 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "3 onComplete");
            }
        };

        mObserver4 = new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "4 onSubscribe");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                LogUtils.d("asdf", "4 onNext : " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "4 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "4 onComplete");
            }
        };
        mObservable2.subscribe(mObserver3);
        mObservable2.subscribe(mObserver4);
    }

    /**
     * 创建观察者
     */
    private void initObserver() {
        mObserver = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                LogUtils.d("asdf", "onNext : " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "onComplete");
            }
        };

        mObserver2 = new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                LogUtils.d("asdf", "2 onSubscribe");
            }

            @Override
            public void onNext(@NonNull String s) {
                LogUtils.d("asdf", "2 onNext : " + s);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                LogUtils.d("asdf", "2 onError");
            }

            @Override
            public void onComplete() {
                LogUtils.d("asdf", "2 onComplete");
            }
        };
    }
}